package com.jason.config;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

/**
 * 自定义配置属性
 * https://www.jianshu.com/p/2447592ca5a9
 */
@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
    private String servers;

    @Value("${spring.kafka.consumer.enable-auto-commit:true}")
    private boolean autoCommit;

    @Value("${spring.kafka.consumer.key-deserializer}")
    private String keyDeserializer;

    @Value("${spring.kafka.consumer.value-deserializer}")
    private String valueDeserializer;



    public Map<String,Object> consumerConfig(){
        Map<String,Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);

        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);  //一次调用poll操作返回的最大记录数，默认500
        return  props;
    }


    /**
     * 自定义DefaultKafkaProducerFactory
     * @return
     */
    @Bean
    public ConsumerFactory consumerFactory(){
        return new DefaultKafkaConsumerFactory(consumerConfig());
    }

    /**
     * 如果在@KafkaListener属性中没有指定 containerFactory 那么Spring Boot 会默认注入 name 为“kafkaListenerContainerFactory”
     * @return
     */
    @Bean("myKafkaListenerContainerFactory")
    public KafkaListenerContainerFactory kafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        //设置并发数
        factory.setConcurrency(1);
        factory.setConsumerFactory(consumerFactory());

        factory.setAutoStartup(true); //自动启动
        //设置批量消费
        factory.setBatchListener(false);
        factory.setMissingTopicsFatal(false);
        factory.getContainerProperties().setPollTimeout(2500);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return  factory;
    }


}
